package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author WEIYUNHUI
 * @date 2023/6/10 11:15
 * <p>
 * 任务槽共享:
 * FLink允许上下游的算子的subTask共享同一个slot.
 * 同一个算子的多个并行度不能共享一个Slot。
 * <p>
 * 如何确定一个作业运行时需要多少个slot?
 * 由当前作业中， 最大并行度的那个算子的并行度来决定。
 * <p>
 * <p>
 * 任务槽共享的好处:
 * 1. Flink 集群所需的 task slot 和作业中使用的最大并行度恰好一样。无需计算程序总共包含多少个 task（具有不同并行度）。
 * 2.资源 容易获得更好的资源利用。如果没有 slot 共享，非密集 subtask（source/map()）将阻塞和密集型 subtask（window） 一样多的
 * <p>
 * 不共享:
 * 通过调整共享组来实现。 slotSharingGroup("a")
 */
public class Flink10_SlotShare {
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 5678);

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //设置并行度
        env.setParallelism(1);

        //env.setRuntimeMode(RuntimeExecutionMode.BATCH)

        //全局禁用算子链
        env.disableOperatorChaining();

        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);

        ds
                .map(x -> x).name("map1").setParallelism(2).slotSharingGroup("a")
                .map(x -> x).name("map2").setParallelism(1).slotSharingGroup("b")
                .map(x -> x).name("map3").setParallelism(1).slotSharingGroup("b")
                .print().setParallelism(1).slotSharingGroup("c");

        // 5. 启动执行
        env.execute();
    }

}